package com.homedo.async.mysql.support.mapper.update;

import com.homedo.async.mysql.support.dml.DMLEnum;
import com.homedo.async.mysql.support.mapper.util.CreateMono4AffectedNumberUtil;
import com.homedo.async.mysql.support.meta.EntityMetaData;
import com.homedo.async.mysql.support.prepare.PrepareObject2Sql;
import com.homedo.async.mysql.support.prepare.PrepareObjectType;
import com.homedo.async.mysql.support.prepare.PrepareRecords2Sql;
import com.homedo.plainly.logger.Logger;
import com.homedo.plainly.logger.LoggerFactory;
import io.vertx.sqlclient.Pool;
import io.vertx.sqlclient.SqlConnection;
import org.reactivestreams.Publisher;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * @author jgshun
 * @date 2020/3/17 10:30 AM
 */
public class UpdateMapperImpl<T, ID> implements UpdateMapperI<T, ID> {
    private static final Logger _LOGGER = LoggerFactory.getLogger(UpdateMapperImpl.class);

    private Pool pool;

    private EntityMetaData<T> entityMetaData;

    public UpdateMapperImpl(Pool pool, EntityMetaData<T> entityMetaData) {
        this.pool = pool;
        this.entityMetaData = entityMetaData;
    }

    @Override
    public Mono<Long> updateByIdSelective(T entity) {
        return this.updateById(entity, false, null);
    }

    @Override
    public Mono<Long> updateByIdsSelective(Iterable<? extends T> entities) {
        return this.updateByIds(entities, false, null);
    }

    @Override
    public Mono<Long> updateByIdsSelective(Publisher<? extends T> entityStream) {
        return this.updateByIds(entityStream, false, null);
    }

    @Override
    public Mono<Long> updateById(T entity) {
        return this.updateById(entity, true, null);
    }

    @Override
    public Mono<Long> updateByIds(Iterable<? extends T> entities) {
        return this.updateByIds(entities, true, null);
    }

    @Override
    public Mono<Long> updateByIds(Publisher<? extends T> entityStream) {
        return this.updateByIds(entityStream, true, null);
    }


    //====================================

    @Override
    public Mono<Long> updateByIdSelective(T entity, SqlConnection connection) {
        return this.updateById(entity, false, connection);
    }

    @Override
    public Mono<Long> updateByIdsSelective(Iterable<? extends T> entities, SqlConnection connection) {
        return this.updateByIds(entities, false, connection);
    }

    @Override
    public Mono<Long> updateByIdsSelective(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.updateByIds(entityStream, false, connection);
    }

    @Override
    public Mono<Long> updateById(T entity, SqlConnection connection) {
        return this.updateById(entity, true, connection);
    }

    @Override
    public Mono<Long> updateByIds(Iterable<? extends T> entities, SqlConnection connection) {
        return this.updateByIds(entities, true, connection);
    }

    @Override
    public Mono<Long> updateByIds(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.updateByIds(entityStream, true, connection);
    }


    private Mono<Long> updateById(T entity, boolean includeNullVals, SqlConnection connection) {
        Assert.notNull(entity, "The given entity must not be null!");

        Object[] prepareSqlArgs = new PrepareObject2Sql(entity, entityMetaData, DMLEnum.UPDATE, PrepareObjectType.ENTITY_RECORD, includeNullVals).prepare();

        return this.createUpdateMono(prepareSqlArgs, connection);
    }

    private Mono<Long> updateByIds(Iterable<? extends T> entities, boolean includeNullVals, SqlConnection connection) {
        Assert.notNull(entities, "The given entities must not be null!");

        Object[] prepareSqlArgs = new PrepareRecords2Sql(entityMetaData, DMLEnum.UPDATE, entities, includeNullVals).prepare();

        return this.createUpdateMono(prepareSqlArgs, connection);
    }

    private Mono<Long> updateByIds(Publisher<? extends T> entityStream, boolean includeNullVals, SqlConnection connection) {
        Assert.notNull(entityStream, "The given Publisher of Entity's must not be null!");

        Mono<Long> updateMono = Mono.create(sink -> {
            Flux.from(entityStream).buffer().subscribe(e -> {
                this.updateByIds(e, includeNullVals, connection).subscribe(res -> {
                    sink.success(res);
                }, error -> sink.error(error));

            }, error -> sink.error(error));
        });

        return updateMono;
    }

    private Mono<Long> createUpdateMono(Object[] prepareSqlArgs, SqlConnection connection) {
        return CreateMono4AffectedNumberUtil.create(prepareSqlArgs, this.pool, connection);
    }

}
